/**
 * Copyright [2019] [LiBo/Alex of copyright liboware@gmail.com ]
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.hyts.rxjava.async;

import rx.Scheduler;
import rx.functions.Action1;
import rx.schedulers.Schedulers;

import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.Observable;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;


/**
 * @project-name:coffice-toolbox
 * @package-name:com.hyts.rxjava.async
 * @author:LiBo/Alex
 * @create-date:2019-08-28 14:51
 * @copyright:libo-alex4java
 * @email:liboware@gmail.com
 * @description: 方法处理操作
 */
public class Async extends AsyncRx<Function, Object> {

    @Override
    public Object batchExecute(Object paramHandler,Function... param) {
        Objects.requireNonNull(param);
        Action1 paramHandlerSubject = (Action1)paramHandler;
        List<Function> list = Arrays.asList(param);
        Iterable paramList = list.parallelStream().map(func -> {
            rx.Observable observable = null;
            observable = rx.Observable.create(paramE -> {
                try {
                    paramE.onNext(param);
                } catch (Exception e) {
                    paramE.onError(e);
                } finally {
                    paramE.onCompleted();
                }
            }).subscribeOn(Schedulers.io());
            return observable;
        }).collect(Collectors.toList());
        rx.Observable.merge(paramList, Runtime.getRuntime().availableProcessors()).toBlocking().forEach((paramHandlerSubject));
        return null;
    }
}
